package cn.ymatrix.mxgateKafkaSDKTOYMatrix;

import cn.ymatrix.logger.MxLogger;
import cn.ymatrix.utils.Config;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerWrapper {
  private static final String TAG = KafkaConsumerWrapper.class.getName();
  private final Logger LOGGER = MxLogger.init(KafkaConsumerWrapper.class);
  private final String topic;
  private final KafkaConsumer<String, String> consumer;
  private long latestOffset;

  private final int consumerNum;

  public KafkaConsumerWrapper(String bootStrapServers, String topic, int part, String groupID) {
    Properties props = new Properties();
    props.put(ConsumerConfig.GROUP_ID_CONFIG, groupID);
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServers);
    props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, Config.getAutoCommitIntervalMs());
    props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, Config.getSessionTimeoutMs());
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, Config.getAutoCommit());
    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, Config.getMaxPollRecords());
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, Config.getOffsetReset());
    props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, Config.getMaxPartitionFetchBytes());
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    // 数据没有序列化使用
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

    this.topic = topic;
    this.consumer = new KafkaConsumer<>(props);
    consumer.subscribe(Collections.singletonList(topic));
    this.consumerNum = part;
  }

  public void consumeLoop(long timeoutMillis, KafkaMsgConsumeCallback callback) {
    while (true) {

      ConsumerRecords<String, String> consumerRecords =
          consumer.poll(Duration.ofMillis(timeoutMillis));
      if (consumerRecords.count() == 0 && consumerRecords.isEmpty()) {
        LOGGER.debug(
            "{} pull 0 records from Kafka with latest offset = {} [{}]",
            TAG,
            this.latestOffset,
            this.consumerNum);
      } else {
        LOGGER.debug(
            "{} pull {} records from Kafka with latest offset = {} [{}]",
            TAG,
            consumerRecords.count(),
            this.latestOffset,
            this.consumerNum);
      }

      consumerRecords.forEach(
          stringStringConsumerRecord -> {
            String value = stringStringConsumerRecord.value();

            KafkaConsumerWrapper.this.latestOffset = stringStringConsumerRecord.offset();
            LOGGER.debug(
                "Consume msg from Topic={} Partition={} Offset={}",
                stringStringConsumerRecord.topic(),
                stringStringConsumerRecord.partition(),
                stringStringConsumerRecord.offset());
            callback.convertMsgValueToObj(value, KafkaConsumerWrapper.this.consumerNum);
          });
    }
  }
}
